package com.adcc.ags.cgw.proxy;

import com.adcc.ags.cgw.configuration.GWFactory;
import com.adcc.ags.cgw.configuration.UserListFactory;
import com.adcc.ags.cgw.entity.msg.CGWMsg;
import com.adcc.ags.cgw.entity.msg.MsgType;
import com.adcc.ags.cgw.entity.msg.TrafficMsg;
import com.adcc.ags.cgw.entity.user.User;
import com.adcc.ags.cgw.utils.Constant;
import com.adcc.utility.log.Log;
import com.adcc.utility.mq.configuration.MQConfiguration;
import com.adcc.utility.mq.configuration.MQConfigurationFactory;
import com.adcc.utility.mq.entity.Message;
import com.adcc.utility.mq.transfer.MQConnectionPool;
import com.adcc.utility.mq.transfer.MQConnectionPoolFactory;
import com.adcc.utility.mq.transfer.MQTransfer;
import com.adcc.utility.mq.transfer.MQTransferFactory;
import com.adcc.utility.mq.transfer.ibm.IBMMQConnectionPool;
import com.google.common.base.Optional;
import com.google.common.base.Strings;
import com.google.common.collect.Maps;
import io.dropwizard.lifecycle.Managed;
import io.dropwizard.setup.Environment;

import java.util.Map;


/**
 * 上行报文发送代理
 */
public class UpLinkDispatchProxy implements Managed {

    // 上行报文处理线程名称
    private static final String THREAD_NAME = "Thread.UpLinkProcessor";

    // 单例对象
    private static UpLinkDispatchProxy instance;

    // GWFactory
    private GWFactory gwFactory;

    // MQConfiguration
    private MQConfiguration configuration;

    // MQConnectionPool
    private MQConnectionPool connectionPool;

    // MQTransfer
    private MQTransfer transfer;

    // 上行处理线程
    private UpLinkProcessor processor;

    /**
     * 构造函数
     */
    public UpLinkDispatchProxy() {

    }

    /**
     * 单例方法
     * @return
     */
    public static UpLinkDispatchProxy getInstance() {
        if (instance == null) {
            instance = new UpLinkDispatchProxy();
        }
        return instance;
    }

    public int getMaxLength() {
        return gwFactory.getUplinkMsgMaxLength();
    }

    /**
     * 建立Managed
     * @param environment
     */
    public void build(Environment environment, GWFactory gwFactory) throws Exception {
        if (environment == null) {
            throw new NullPointerException("environment is null");
        }
        if (gwFactory == null) {
            throw new NullPointerException("gwFactory is null");
        }
        this.gwFactory = gwFactory;
        environment.lifecycle().manage(this);
    }

    /**
     * 发送下行报文
     * @param qName
     * @param msg
     * @param type
     * @param sender
     * @throws Exception
     */
    public void sendDownlinkMsg(String qName, byte[] msg, MsgType type, String sender) throws Exception{
        try {
            Map<String, Object> head = Maps.newHashMapWithExpectedSize(2);
            head.put(Constant.MSG_HEAD_TYPE, type.toString());
            head.put(Constant.MSG_HEAD_SENDER, sender);
            Message message = new Message(head, msg);
            transfer.sendQueue(qName, message);
        } catch (Exception ex) {
            Log.error(UpLinkDispatchProxy.class.getName(),"sendDownlinkMsg() error",ex);
            throw ex;
        }
    }

    /**
     * 接收上行报文
     * @param user
     * @return
     */
    public Optional<CGWMsg> receiveUplinkMsg(User user){
        try{

            // 读取上行报文
            Message message = transfer.receiveQueue(user.getSendQueue());
            if(message != null){
                Log.info(UpLinkDispatchProxy.class.getName(), "receive uplink message\r\n" + message.toString());
                CGWMsg cgwMsg = new CGWMsg(message);

                // 取得报文类型
                String strType = (String)message.getHead().get(Constant.MSG_HEAD_TYPE);
                if(Strings.isNullOrEmpty(strType)){
                    Log.error(UpLinkDispatchProxy.class.getName(),"uplink message head type missing");
                    return Optional.absent();
                }
                if(MsgType.UPLINK.toString().equals(strType) || MsgType.GROUND.toString().equals(strType)){
                    cgwMsg.setType(MsgType.valueOf(strType));
                }else{
                    Log.error(UpLinkDispatchProxy.class.getName(),"uplink message head type " + strType + " is invalid");
                    return Optional.absent();
                }

                // 取得发送方ID
                String strSender = (String)message.getHead().get(Constant.MSG_HEAD_SENDER);
                if(!Strings.isNullOrEmpty(strSender)){
                    cgwMsg.setSender(strSender);
                    if(strSender.equals(user.getId())){
                        cgwMsg.setUser(user);
                    }else{
                        for(User u : UserListFactory.getInstance().getUserList()){
                            if(strSender.equals(u.getId())){
                                cgwMsg.setUser(u);
                                break;
                            }
                        }
                    }
                }else{
                    Log.error(UpLinkDispatchProxy.class.getName(),"uplink message head sender missing");
                    return Optional.absent();
                }

                String strReceiver = (String)message.getHead().get(Constant.MSG_HEAD_RECEIVER);
                if(!Strings.isNullOrEmpty(strReceiver)){
                    cgwMsg.setReceiver(strReceiver);
                }
                return Optional.of(cgwMsg);
            }else{
                return Optional.absent();
            }
        }catch (Exception ex){
            Log.error(UpLinkDispatchProxy.class.getName(),"receiveUplinkMsg() error",ex);
            return Optional.absent();
        }
    }

    @Override
    public void start() {
        try {

            // 设置MQ连接配置
            configuration = MQConfigurationFactory.getInstance().createIBMMQConfiguration(gwFactory.getMqFactory().getHost(),
                    gwFactory.getMqFactory().getPort(),
                    gwFactory.getMqFactory().getQueueManager(),
                    gwFactory.getMqFactory().getChannel());

            // 设置MQ连接池配置
            if(gwFactory.getMqFactory().getMaxConnections() > 0 && gwFactory.getMqFactory().getMaxIdelConnections() > 0){
                connectionPool = MQConnectionPoolFactory.getInstance().createIBMMQConnectionPool();
                ((IBMMQConnectionPool)connectionPool).setActiveMode(gwFactory.getMqFactory().getActiveMode());
                ((IBMMQConnectionPool)connectionPool).setTimeout(gwFactory.getMqFactory().getTimeout());
                ((IBMMQConnectionPool)connectionPool).setMaxConnections(gwFactory.getMqFactory().getMaxConnections());
                ((IBMMQConnectionPool)connectionPool).setMaxIdelConnections(gwFactory.getMqFactory().getMaxIdelConnections());
                connectionPool.init(configuration);
            }

            // 创建MQTransfer
            transfer = MQTransferFactory.getInstance().createIBMMQTransfer("Transfer.GW",configuration,connectionPool);

            // 启动上行处理线程
            processor = new UpLinkProcessor();
            new Thread(processor, THREAD_NAME).start();
        } catch (Exception ex) {
            Log.error(UpLinkDispatchProxy.class.getName(), "start() error", ex);
        }
    }

    @Override
    public void stop() throws Exception {
        try {

            // 销毁MQTransfer
            MQTransferFactory.getInstance().disposeMQTransfer("Transfer.GW");

            // 关闭上行处理线程
            if (processor != null) {
                processor.close();
            }
        } catch (Exception ex) {
            Log.error(UpLinkDispatchProxy.class.getName(), "stop() error", ex);
        }
    }

    /**
     * 上行处理线程
     */
    class UpLinkProcessor implements Runnable {

        // 线程启动标识
        private boolean isStarted;

        /**
         * 关闭线程
         */
        public void close(){
            isStarted = false;
        }

        @Override
        public void run() {
            try {
                Log.info(UpLinkProcessor.class.getName(), "UpLinkProcessor start...");
                isStarted = true;
                while (isStarted) {
                    try {
                        if(UserListFactory.getInstance().getUserList().size() > 0){
                            for(User user : UserListFactory.getInstance().getUserList()){
                                Optional<CGWMsg> optional = receiveUplinkMsg(user);
                                if(optional.isPresent()){
                                    CGWMsg cgwMsg = optional.get();

                                    // 执行MsgProxy上行报文业务逻辑链
                                    MsgProxy.getInstance().getMsgVerifyActor().tell(cgwMsg, MsgProxy.getInstance().getMsgVerifyActor());
                                    MsgProxy.getInstance().getMsgSaveActor().tell(new TrafficMsg(cgwMsg),MsgProxy.getInstance().getMsgSaveActor());
                                }
                            }
                        }
                        Thread.sleep(50);
                    } catch (Exception ex) {
                        Log.error(UpLinkProcessor.class.getName(), "process message error", ex);
                    }
                }
            }catch (Exception ex){
                Log.error(UpLinkProcessor.class.getName(), "run() error",ex);
            }
        }
    }
}
